package com.amap.search.service;

import com.alibaba.otter.canal.protocol.CanalEntry;
import com.amap.search.entity.db2022.AddressCode;
import com.amap.search.entity.db2022.FileRecord;
import com.amap.search.entity.db2022.PoiRecord;
import com.amap.search.entity.es.PoiRecordEs;
import com.amap.search.mapper.db2022.AddressCodeMapper;
import com.amap.search.mapper.db2022.FileRecordMapper;
import com.amap.search.mapper.db2022.PoiRecordMapper;
import com.amap.search.mapper.es.PoiRecordDao;
import com.amap.search.mapper.es.WeatherRecordDao;
import com.baomidou.mybatisplus.mapper.EntityWrapper;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.collections4.CollectionUtils;
import org.apache.commons.lang3.StringUtils;
import org.springframework.stereotype.Service;

import javax.annotation.Resource;
import java.math.BigDecimal;
import java.text.ParseException;
import java.text.SimpleDateFormat;
import java.util.ArrayList;
import java.util.Date;
import java.util.List;
import java.util.stream.Collectors;

@Service
@Slf4j
public class ElasticSearchOperatorService {

    @Resource
    private PoiRecordDao poiRecordDao;

    @Resource
    private TransformService transformService;

    @Resource
    private PoiRecordMapper poiRecordMapper;

    ArrayList<PoiRecordEs> poiRecordEsList = new ArrayList<>(64);

    /**
     * 从canal同步增量数据到ES
     * @param rowData
     */
    public void insertFromCanal(CanalEntry.RowData rowData) {
        if (CollectionUtils.isEmpty(rowData.getAfterColumnsList())) return;
        PoiRecordEs record = transformService.getEsPoiFromCanal(rowData);
        addElement(record);
    }

    public synchronized void addElement(PoiRecordEs poiRecordEs) {
        this.poiRecordEsList.add(poiRecordEs);
        if (poiRecordEsList.size() >= 20) {
            poiRecordDao.saveAll(poiRecordEsList);
            poiRecordEsList.clear();
        }
    }

    public synchronized void pushElement(){
        poiRecordDao.saveAll(poiRecordEsList);
        poiRecordEsList.clear();
    }

    public void delete(CanalEntry.RowData rowData) {
        PoiRecordEs poiRecordEs = new PoiRecordEs();
        for (CanalEntry.Column column : rowData.getBeforeColumnsList()) {
            if ("id".equals(column.getName())) {
                poiRecordEs.setId(Long.parseLong(column.getValue()));
                poiRecordDao.delete(poiRecordEs);
                return;
            }
        }
    }

    /**
     * 手动将数据库中全量数据同步至ES
     */
    public void synFromMysqlToEs(){
        int page = 0;
        do {
            List<PoiRecord> poiRecordList = poiRecordMapper.queryAllxRecordByPage(page, 500);
            List<PoiRecordEs> poiRecordEsList = transformService.getEsPoiFromMysql(poiRecordList);
            if(!CollectionUtils.isEmpty(poiRecordEsList)){
                poiRecordDao.saveAll(poiRecordEsList);
                page++;
            }
        } while (CollectionUtils.isEmpty(poiRecordEsList));
    }

}
